package com.demo.rocketmq.sender;

import com.demo.rocketmq.config.RocketMqDelayLevelEnum;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.List;


/**
 * 发送同步消息
 */
@Slf4j
@Component
public class RocketMqSender {

  @Autowired
  private DefaultMQProducer defaultMQProducer;

  private static final MessageQueueSelector selector = new MessageQueueSelector() {
    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object regionKey) {
      int hashCode = regionKey.hashCode();
      int regionKeyValue = 0;
      if (regionKey instanceof String) {
        regionKeyValue = Math.abs(hashCode);
      } else if (regionKey instanceof Integer) {
        regionKeyValue = (Integer) regionKey;
      } else {
        regionKeyValue = Math.abs(hashCode);
      }
      int index = regionKeyValue % mqs.size();
      return mqs.get(index);
    }
  };


  /**
   * 发送同步消息
   *
   * @param topic
   * @param keys    业务唯一标识作为幂等处理，https://help.aliyun.com/document_detail/44397.htm?spm=a2c4g.11186623.2.13.618b7907ZyouGT#concept-2047147
   * @param message
   * @return
   */
  public SendResult send(String topic, String keys, String message) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      return defaultMQProducer.send(msg);
    } catch (Exception e) {
      log.error("MQ发送消息异常，topic = {}，keys = {}，message = {}，e = {}", topic, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }

  public SendResult sendWithTag(String topic, String tag, String keys, String message) {
    try {
      Message msg = new Message(topic, tag, keys, toByte(message));
      return defaultMQProducer.send(msg);
    } catch (Exception e) {
      log.error("MQ发送异步消息异常，topic = {}，tag = {}, keys = {}，message = {}，e = {}", topic, tag, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }

  public SendResult sendWithSelector(String topic, String keys, String message, String regionKey) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      return defaultMQProducer.send(msg, selector, regionKey);
    } catch (Exception e) {
      log.error("MQ发送消息异常，topic = {}，keys = {}，message = {}，e = {}", topic, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }

  public SendResult sendWithTagAndSelector(String topic, String tag, String keys, String message, String regionKey) {
    try {
      Message msg = new Message(topic, tag, keys, toByte(message));
      return defaultMQProducer.send(msg, selector, regionKey);
    } catch (Exception e) {
      log.error("MQ发送异步消息异常，topic = {}，tag = {}, keys = {}，message = {}，e = {}", topic, tag, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }

  public SendResult sendDelay(String topic, String keys, String message, RocketMqDelayLevelEnum rocketMqDelayLevelEnum) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      msg.setDelayTimeLevel(rocketMqDelayLevelEnum.getLevel());
      return defaultMQProducer.send(msg);
    } catch (Exception e) {
      log.error("MQ发送消息异常，topic = {}，keys = {}，message = {}，e = {}", topic, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }

  public SendResult sendDelayWithTag(String topic, String tag, String keys, String message, RocketMqDelayLevelEnum rocketMqDelayLevelEnum) {
    try {
      Message msg = new Message(topic, tag, keys, toByte(message));
      msg.setDelayTimeLevel(rocketMqDelayLevelEnum.getLevel());
      return defaultMQProducer.send(msg);
    } catch (Exception e) {
      log.error("MQ发送消息异常，topic = {}，tag = {}，keys = {}，message = {}，e = {}", topic, tag, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }

  public SendResult sendDelayBySec(String topic, String keys, String message, long sec) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      msg.setDelayTimeSec(sec);
      return defaultMQProducer.send(msg);
    } catch (Exception e) {
      log.error("MQ发送消息异常，topic = {}，keys = {}，message = {}，e = {}", topic, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }

  public SendResult sendDelayByMs(String topic, String keys, String message, long timeMs) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      msg.setDelayTimeMs(timeMs);
      return defaultMQProducer.send(msg);
    } catch (Exception e) {
      log.error("MQ发送消息异常，topic = {}，keys = {}，message = {}，e = {}", topic, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }

  public SendResult sendDelayByDeliverTimeMs(String topic, String keys, String message, long timeMs) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      msg.setDeliverTimeMs(timeMs);
      return defaultMQProducer.send(msg);
    } catch (Exception e) {
      log.error("MQ发送消息异常，topic = {}，keys = {}，message = {}，e = {}", topic, keys, message, ExceptionUtils.getStackTrace(e));
    }
    return null;
  }


  /**
   * 发送异步消息
   *
   * @param topic
   * @param message
   * @return
   */
  public void sendAsync(String topic, String keys, String message) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      defaultMQProducer.send(msg, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
          if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送异步消息成功，sendResult = {}", sendResult);
          } else {
            log.error("MQ发送异步消息失败，sendResult = {}", sendResult);
          }
        }

        @Override
        public void onException(Throwable e) {
          log.error("MQ发送异步消息异常，topic = {}，message = {}，e = {}", topic, message, ExceptionUtils.getStackTrace(e));
        }
      });
    } catch (Exception e) {
      log.error("MQ发送异步消息异常，topic = {}，message = {}，e = {}", topic, message, ExceptionUtils.getStackTrace(e));
    }
  }

  public void sendAsyncWithTag(String topic, String tag, String keys, String message) {
    try {
      Message msg = new Message(topic, tag, keys, toByte(message));
      defaultMQProducer.send(msg, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
          if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送异步消息成功，sendResult = {}", sendResult);
          } else {
            log.error("MQ发送异步消息失败，sendResult = {}", sendResult);
          }
        }

        @Override
        public void onException(Throwable e) {
          log.error("MQ发送异步消息异常，topic = {}，tag = {}, message = {}，onException = {}", topic, tag, message, ExceptionUtils.getStackTrace(e));
        }
      });
    } catch (Exception e) {
      log.error("MQ发送异步消息异常，topic = {}，tag = {}, message = {}，e = {}", topic, tag, message, ExceptionUtils.getStackTrace(e));
    }
  }

  public void sendAsyncWithSelector(String topic, String keys, String message, String regionKey) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      defaultMQProducer.send(msg, selector, regionKey, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
          if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送异步消息成功，sendResult = {}", sendResult);
          } else {
            log.error("MQ发送异步消息失败，sendResult = {}", sendResult);
          }
        }

        @Override
        public void onException(Throwable e) {
          log.error("MQ发送异步消息异常，topic = {}，message = {}，onException = {}", topic, message, ExceptionUtils.getStackTrace(e));
        }
      });
    } catch (Exception e) {
      log.error("MQ发送异步消息异常，topic = {}，message = {}，e = {}", topic, message, ExceptionUtils.getStackTrace(e));
    }
  }

  public void sendAsyncWithTagAndSelector(String topic, String tag, String keys, String message, String regionKey) {
    try {
      Message msg = new Message(topic, tag, keys, toByte(message));
      defaultMQProducer.send(msg, selector, regionKey, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
          if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送异步消息成功，sendResult = {}", sendResult);
          } else {
            log.error("MQ发送异步消息失败，sendResult = {}", sendResult);
          }
        }

        @Override
        public void onException(Throwable e) {
          log.error("MQ发送异步消息异常，topic = {}，tag = {}, message = {}，onException = {}", topic, tag, message, ExceptionUtils.getStackTrace(e));
        }
      });
    } catch (Exception e) {
      log.error("MQ发送异步消息异常，topic = {}，tag = {}, message = {}，e = {}", topic, tag, message, ExceptionUtils.getStackTrace(e));
    }
  }

  public void sendAsyncDelay(String topic, String keys, String message, RocketMqDelayLevelEnum rocketMqDelayLevelEnum) {
    try {
      Message msg = new Message(topic, null, keys, toByte(message));
      msg.setDelayTimeLevel(rocketMqDelayLevelEnum.getLevel());
      defaultMQProducer.send(msg, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
          if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送异步消息成功，sendResult = {}", sendResult);
          } else {
            log.error("MQ发送异步消息失败，sendResult = {}", sendResult);
          }
        }

        @Override
        public void onException(Throwable e) {
          log.error("MQ发送异步消息异常，topic = {}，keys = {}，message = {}，rocketMqDelayLevelEnum = {}，onException = {}", topic, keys, message, rocketMqDelayLevelEnum.getLevel(), ExceptionUtils.getStackTrace(e));
        }
      });
    } catch (Exception e) {
      log.error("MQ发送异步消息异常，topic = {}，keys = {}，message = {}，rocketMqDelayLevelEnum = {}，e = {}", topic, keys, message, rocketMqDelayLevelEnum.getLevel(), ExceptionUtils.getStackTrace(e));
    }
  }

  public void sendAsyncDelayWithTag(String topic, String tag, String keys, String message, RocketMqDelayLevelEnum rocketMqDelayLevelEnum) {
    try {
      Message msg = new Message(topic, tag, keys, toByte(message));
      msg.setDelayTimeLevel(rocketMqDelayLevelEnum.getLevel());
      defaultMQProducer.send(msg, new SendCallback() {
        @Override
        public void onSuccess(SendResult sendResult) {
          if (sendResult.getSendStatus().equals(SendStatus.SEND_OK)) {
            log.info("MQ发送异步消息成功，sendResult = {}", sendResult);
          } else {
            log.error("MQ发送异步消息失败，sendResult = {}", sendResult);
          }
        }

        @Override
        public void onException(Throwable e) {
          log.error("MQ发送异步消息异常，topic = {}，tag = {}，keys = {}，message = {}，rocketMqDelayLevelEnum = {}，onException = {}", topic, tag, keys, message, rocketMqDelayLevelEnum.getLevel(), ExceptionUtils.getStackTrace(e));
        }
      });
    } catch (Exception e) {
      log.error("MQ发送异步消息异常，topic = {}，tag = {}，keys = {}，message = {}，rocketMqDelayLevelEnum = {}，e = {}", topic, tag, keys, message, rocketMqDelayLevelEnum.getLevel(), ExceptionUtils.getStackTrace(e));
    }
  }

  private byte[] toByte(String body) {
    return body.getBytes(StandardCharsets.UTF_8);
  }

}
